package com.tang.state;

import com.tang.bean.WaterSensor;
import com.tang.functions.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 计算每种传感器的水位和
 *
 * @author tang
 * @since 2023/7/11 11:00
 */
public class KeyedReduceStateDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.socketTextStream("192.168.70.141", 7777)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000L)
                ).keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    // 1. 定义ListState
                    private ReducingState<Integer> reducingState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        // 2. 在Open方法中初始化状态
                        reducingState = getRuntimeContext().getReducingState(
                                new ReducingStateDescriptor<>("reducingState", Integer::sum, Types.INT()));
                        /*reducingState = getRuntimeContext().getReducingState(
                                new ReducingStateDescriptor<>("reducingState", new ReduceFunction<Integer>() {
                                    @Override
                                    public Integer reduce(Integer value1, Integer value2) throws Exception {
                                        return value1 + value2;
                                    }
                                }, Types.INT()));*/
                    }

                    @Override
                    public void processElement(WaterSensor waterSensor,
                                               KeyedProcessFunction<String, WaterSensor, String>.Context ctx,
                                               Collector<String> out) throws Exception {
                        reducingState.add(waterSensor.getVc());
                        Integer vcSum = reducingState.get();
                        out.collect("传感器id为" + waterSensor.getId() + ", 水位值总和 = " + vcSum);
//                                reducingState.get();   // 对本组的Reducing状态，获取结果
//                                reducingState.add();   // 对本组的Reducing状态，添加数据
//                                reducingState.clear(); // 对本组的Reducing状态，清空数据
                    }

                })
                .print();

        env.execute();
    }

}
